package com.huonan.component.kafka.config;

import lombok.Getter;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 *
 */
@Configuration
@EnableKafka
//@ConditionalOnProperty
@Getter
public class KafkaProducerConfig {

    /**
     * 主机地址
     */
    @Value("${kafka.server-host:10.11.110.78:9092}")
    private String bootstrapServers;
//
//    /**
//     * sasl 认证账号
//     */
//    @Value("${iot.kafka.sasl.username:admin}")
//    private String userName;
//
//    /**
//     * sasl 密码
//     */
//    @Value("${iot.kafka.sasl.password:iot@2021}")
//    private String password;

    /**
     * 重试次数 默认值0
     */
    @Value("${iot.kafka.producer.retries:10}")
    private Integer retries;

    /**
     * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
     * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
     * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
     */
    @Value("${iot.kafka.producer.acks:1}")
    private String acks;

    /**
     * 指定缓存的大小，生产者缓存每个分区未发送的消息。默认 16384
     */
    @Value("${iot.kafka.producer.batch-size:1024}")
    private Integer batchSize;

    /**
     * 生产者发送请求之前等待一段时间，设置等待时间是希望更多地消息填补到未满的批中。 默认 30
     */
    @Value("${iot.kafka.producer.properties.linger.ms:30}")
    private Integer lingerMs;

    /**
     * 通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的 默认32m
     */
    @Value("${iot.kafka.producer.buffer-memory:33554432}")
    private Integer bufferMemory;

    /**
     * 生产者配置
     *
     * @return 配置
     */
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>(11);
        //kafka server地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        /*
         * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
         * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
         * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
         */
        props.put(ProducerConfig.ACKS_CONFIG, acks);
        //消息发送失败重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        //去缓冲区中一次拉16k的数据，发送到broker
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        // 批量发送，延迟为30毫秒，如果30ms内凑不够batch则强制发送，提高并发
        props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
        //设置缓存区大小 32m
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        //key序列化器选择
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //value序列化器选择
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //分区策略配置
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
        //设置sasl认证
//        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
//        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
//        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required " +
//                "username='" + userName + "' password='" + password + "';");
        return props;
    }

    /**
     * 不使用spring boot的KafkaAutoConfiguration默认方式创建的DefaultKafkaProducerFactory，重新定义
     * @return
     */
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    /**
     * Producer Template 配置
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }
}
